package com.wfg.flink.connector.mongodb.schema;

import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.JSONReader;
import com.wfg.flink.connector.mongodb.model.WellCastingInfo;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.connector.mongodb.source.reader.deserializer.MongoDeserializationSchema;
import org.bson.BsonDocument;

import java.util.Date;

/**
 * @author wfg
 */
@Slf4j
public class WellCastingInfoDeserializationSchema implements MongoDeserializationSchema<WellCastingInfo> {
    @Override
    public WellCastingInfo deserialize(BsonDocument bsonDocument) {
        WellCastingInfo rs = null;
        try {
            JSONObject obj = JSONObject.parseObject(bsonDocument.toJson());
            obj.remove("_id");
            obj.remove("time");
            obj.remove("_class");
            rs = obj.to(WellCastingInfo.class, JSONReader.Feature.IgnoreAutoTypeNotMatch);
            if (bsonDocument.getObjectId("_id") != null) {
                rs.setId(bsonDocument.getObjectId("_id").getValue().toString());
            }
            if (bsonDocument.get("time") != null) {
                rs.setTime(new Date(bsonDocument.getDateTime("time").getValue()));
            }
        } catch (Exception e) {
            log.error("数据格式错误：{}:{}", bsonDocument.toJson(), e);
        }
        return rs;
    }

    @Override
    public TypeInformation<WellCastingInfo> getProducedType() {
        return TypeExtractor.getForClass(WellCastingInfo.class);
    }

}
